package org.eleorc.flow;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Flow;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.function.Consumer;

public class FlowDemo00 {

    public static class OneShotPublisher implements Flow.Publisher<Boolean>{

        private final ExecutorService executor = ForkJoinPool.commonPool(); // daemon-based
        private boolean subscribed; // true after first subscribe



        @Override
        public synchronized void subscribe(Flow.Subscriber<? super Boolean> subscriber) {
            if (subscribed){
                subscriber.onError(new IllegalStateException()); // only one allowed
            }else {
                subscribed = true;
                subscriber.onSubscribe(new OneShotSubscription(subscriber, executor));
            }
        }

        static class OneShotSubscription implements Flow.Subscription {
            private final Flow.Subscriber<? super Boolean> subscriber;
            private final ExecutorService executor;
            private Future<?> future; // to allow cancellation
            private boolean completed;

            OneShotSubscription(Flow.Subscriber<? super Boolean> subscriber, ExecutorService executor) {
                this.subscriber = subscriber;
                this.executor = executor;
            }

            @Override
            public synchronized void request(long n) {
                if (!completed){
                    completed = true;
                    if (n <= 0) {
                        IllegalArgumentException ex = new IllegalArgumentException();
                        executor.execute(() -> subscriber.onError(ex));
                    }else {
                        future = executor.submit(() -> {
                            subscriber.onNext(Boolean.TRUE);
                            subscriber.onComplete();
                        });
                    }
                }
            }

            @Override
            public void cancel() {
                completed = true;
                if (future != null) future.cancel(false);
            }
        }
    }


    public static class SampleSubscriber<T> implements Flow.Subscriber<T>{

        final Consumer<? super T> consumer;
        Flow.Subscription subscription;
        final long bufferSize;
        long count;

        public SampleSubscriber(Consumer<? super T> consumer, long bufferSize) {
            this.consumer = consumer;
            this.bufferSize = bufferSize;
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            long initialRequestSize = bufferSize;
            count = bufferSize - bufferSize / 2; // re-request when half consumed
            (this.subscription = subscription).request(initialRequestSize);
        }

        @Override
        public void onNext(T item) {
            if (--count <= 0)
                subscription.request(count = bufferSize - bufferSize / 2);
            consumer.accept(item);
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println(Thread.currentThread().getName()+":"+this+" onError");
            throwable.printStackTrace();
        }

        @Override
        public void onComplete() {
            System.out.println(Thread.currentThread().getName()+":"+this+"=>complete");
            subscription.cancel();
        }
    }


    public static class  UnboundedSubscriber<T> implements Flow.Subscriber<T> {

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            subscription.request(Long.MAX_VALUE); // effectively unbounded
        }

        @Override
        public void onNext(T item) {
            use(item);
        }

        @Override
        public void onError(Throwable throwable) {
            throwable.printStackTrace();
        }

        @Override
        public void onComplete() {
            System.out.println(this+":complete");
        }

        void use(T item) {
            System.out.println(item);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        OneShotPublisher oneShotPublisher = new OneShotPublisher();
        SampleSubscriber<Boolean> subscriber = new SampleSubscriber<Boolean>(item-> System.out.println(Thread.currentThread().getName()+":"+item),10);
        oneShotPublisher.subscribe(subscriber);


        Thread.sleep(3000);
    }

}
